package wordcount;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

public class WordCountTopology {
    public static void main(String[] args) throws Exception {
        // 实例化对象
        SentenceSpout sentenceSpout = new SentenceSpout();
        SentenceBolt sentenceBolt1 = new SentenceBolt();
        WordCountBolt wordCountBolt2 = new WordCountBolt();
        ReportBolt reportBolt3 = new ReportBolt();

        // 拓扑构建者
        TopologyBuilder builder = new TopologyBuilder();

        // sentenceSpout --> sentenceBolt1 --> wordCountBolt2 --> reportBolt3
        // 如果线程个数和task个数相等，可以省略setNumTasks()
        builder.setSpout("sentenceSpout", sentenceSpout, 2);
        // shuffle  随机的数据流分组
        builder.setBolt("sentenceBolt1", sentenceBolt1, 2).setNumTasks(4).shuffleGrouping("sentenceSpout");
        // fields   字段数据流分区，确保相同的字段进入到同一个bolt
        builder.setBolt("wordCountBolt2", wordCountBolt2, 4).fieldsGrouping("sentenceBolt1", new Fields("word"));
        // global   全局分组，所有的数据流都进入到一个bolt
        builder.setBolt("reportBolt3", reportBolt3).globalGrouping("wordCountBolt2");

        StormTopology topology = builder.createTopology();

        Config config = new Config();
        // 设置当前拓扑的进程并发度
        config.setNumWorkers(2);

        if (args != null && args.length > 0) {
            // 集群模式
            // storm jar storm_word_count.jar wordcount.WordCountTopology 123
            // storm kill 123
            StormSubmitter.submitTopology(args[0], config, topology);

        } else {
            // 本地集群模式
            // storm local storm_word_count.jar wordcount.WordCountTopology
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology("wordCountTopology", config, topology);
        }
    }
}

// help: storm rebalance -h
// storm jar ch2.jar wordcount.WordCountTopology 12345
// storm rebalance 12345 -n 4 -e sentenceBolt1=4
// then waiting 30 second

// storm rebalance 12345 -n 4 -e sentenceBolt1=2 wordCountBolt2=4

// 一个worker有一个acker